package com.study.flink.source;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import com.study.flink.entity.does.HumidityDO;
import com.study.flink.entity.does.TemperatureDO;

public class TemperatureSource implements SourceFunction<TemperatureDO> {
    private static final long serialVersionUID = 1L;

    private FileInputStream stream;
    private BufferedReader reader;

    private final int watermarkDelayMSecs;
    private final int maxDelayMsecs;
    private final int servingSpeed;

    public TemperatureSource() {
        watermarkDelayMSecs = 24 * 60 * 60 * 1000;
        maxDelayMsecs = 10;
        servingSpeed = 10;
    }

    @Override
    public void run(SourceContext<TemperatureDO> ctx) throws Exception {
        stream = new FileInputStream(HumidityDO.FIEL_PATH);
        reader = new BufferedReader(new InputStreamReader(stream));
        generateLimitedStream(ctx);
    }

    private void generateLimitedStream(SourceContext<TemperatureDO> ctx)
        throws IOException, InterruptedException, ParseException {
        long timestamp = new Date().getTime();
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-mm-dd HH:MM:ss");
        long nextWatermark = 0;

        while (reader.ready()) {
            String line = reader.readLine();
            String[] fields = line.split(",");
            TemperatureDO temperatureDODO = new TemperatureDO();
            temperatureDODO.setTimestamp(fields[0].trim());
            temperatureDODO.setTemperature(fields[1].trim());
            long eventTime = dateFormat.parse(temperatureDODO.getTimestamp()).getTime();

            if (eventTime > nextWatermark) {
                if (nextWatermark == 0) {
                    nextWatermark = eventTime;
                }

                nextWatermark = nextWatermark + watermarkDelayMSecs;
                ctx.emitWatermark(new Watermark(nextWatermark));
            }

            ctx.collectWithTimestamp(temperatureDODO, eventTime);
            Thread.sleep(servingSpeed);
        }

    }

    @Override
    public void cancel() {
        try {
            stream.close();
        }
        catch (IOException e) {
            e.printStackTrace();
        }

        try {
            reader.close();
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }

}
